package com.zhang.storm.pv;

import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Tuple;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;

public class PvSumBolt extends BaseRichBolt {

    private Map<Long, Integer> map = new HashMap<>();

    @Override
    public void prepare(Map map, TopologyContext topologyContext, OutputCollector outputCollector) {

    }

    @Override
    public void execute(Tuple input) {
        Long threadId = input.getLong(0);
        Integer pvnum = input.getInteger(1);
        map.put(threadId, pvnum);
        Iterator<Integer> iterator = map.values().iterator();
        int sumnum = 0;
        while (iterator.hasNext()) {
            sumnum += iterator.next();
        }
        System.err.println("sumnum=" + sumnum);
    }

    @Override
    public void declareOutputFields(OutputFieldsDeclarer declarer) {

    }
}
